package org.alis.ap.rpc.client.handler;

import lombok.extern.slf4j.Slf4j;
import org.alis.ap.rpc.Request;
import org.alis.ap.rpc.Response;
import org.alis.ap.rpc.ServiceDescriptor;
import org.alis.ap.rpc.client.selector.TransportSelector;
import org.alis.ap.rpc.codec.Decoder;
import org.alis.ap.rpc.codec.Encoder;
import org.alis.ap.rpc.transport.TransportClient;
import org.alis.rpc.common.exception.NetWorkException;
import org.apache.commons.io.IOUtils;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Objects;

/**
 * RemoteInvoker desc: 调用远程服务的代理类
 *
 * @author luochuan
 * @date 2020/10/21 14:26
 */
@Slf4j
public class RemoteInvoker<T> implements InvocationHandler {
    private Encoder encoder;

    private Decoder decoder;

    private TransportSelector selector;

    private Class<T> clazz;

    public RemoteInvoker(Class<T> clazz, Encoder encoder, Decoder decoder, TransportSelector selector) {
        this.decoder = decoder;
        this.encoder = encoder;
        this.selector = selector;
        this.clazz = clazz;
    }

    @Override
    public T invoke(Object proxy, Method method, Object[] args) {

        Request request = new Request();
        request.setServiceDescriptor(ServiceDescriptor.from(clazz, method));
        request.setParameter(args);
        Response<T> resp = invokeRemote(request);
        if (Objects.isNull(resp) || resp.getCode() != 0) {
            throw new NetWorkException("remote invoker failed " + resp);
        }
        return resp.getData();
    }

    private Response<T> invokeRemote(Request request) {
        TransportClient client = null;
        Response<T> response;
        try {
            client = selector.select();
            byte[] encode = encoder.encode(request);
            InputStream write = client.write(new ByteArrayInputStream(encode));
            byte[] bytes = IOUtils.readFully(write, write.available());
            response = (Response<T>) decoder.decode(bytes,Response.class);
        } catch (IOException e) {
            log.warn(e.getMessage(),e);
            response = new Response<>();
            response.setCode(1);
            response.setMessage("ap rpc client get err:" +e.getClass()+":"+e.getMessage());
        } finally {
            if (Objects.nonNull(client)) {
                selector.release(client);
            }
        }

        return response;

    }
}
